package com.atguigu.gmall.realtime.app.dws;

import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.JSONObject;
import com.atguigu.gmall.realtime.app.dws.func.AbsDimAsyncFuntion;
import com.atguigu.gmall.realtime.app.dws.func.DimAsyncFuntion;
import com.atguigu.gmall.realtime.bean.TradeSkuOrderBean;
import com.atguigu.gmall.realtime.bean.UserLoginBean;
import com.atguigu.gmall.realtime.util.ClickhouseUtil;
import com.atguigu.gmall.realtime.util.DateFormatUtil;
import com.atguigu.gmall.realtime.util.DimUtil;
import com.atguigu.gmall.realtime.util.MyKafkaUtil;
import org.apache.flink.api.common.eventtime.SerializableTimestampAssigner;
import org.apache.flink.api.common.eventtime.WatermarkStrategy;
import org.apache.flink.api.common.functions.MapFunction;
import org.apache.flink.api.common.functions.ReduceFunction;
import org.apache.flink.api.common.state.ValueState;
import org.apache.flink.api.common.state.ValueStateDescriptor;
import org.apache.flink.api.java.functions.KeySelector;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.datastream.AsyncDataStream;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.datastream.KeyedStream;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.KeyedProcessFunction;
import org.apache.flink.streaming.api.functions.ProcessFunction;
import org.apache.flink.streaming.api.functions.windowing.WindowFunction;
import org.apache.flink.streaming.api.windowing.assigners.TumblingEventTimeWindows;
import org.apache.flink.streaming.api.windowing.time.Time;
import org.apache.flink.streaming.api.windowing.windows.TimeWindow;
import org.apache.flink.util.Collector;

import java.math.BigDecimal;
import java.time.Duration;
import java.util.HashMap;
import java.util.Map;
import java.util.concurrent.TimeUnit;

public class DwsTradeSkuOrderWindow {


    public static void main(String[] args) throws Exception {

        //0 环境
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(4);

        //1数据源：dwd_traffic_page_log 用户的页面浏览
        //1  获取数据流
        String sourceTopic="dwd_trade_order_detail";
        String groupId="dws_trade_sku_order_window";
        DataStreamSource<String> kafkaStream = env.addSource(MyKafkaUtil.getKafkaConsumer(sourceTopic, groupId));
        SingleOutputStreamOperator<JSONObject> jsonObjStream = kafkaStream.map(JSON::parseObject);



        //每一笔数据到达 ，都通过状态记录，如果有新的更完整的数据到达，从状态中取得旧数据，生成反向的负值 ，再把新数据放进状态
        KeyedStream<JSONObject, String> detailIdKeyedStream = jsonObjStream.keyBy(new KeySelector<JSONObject, String>() {
            @Override
            public String getKey(JSONObject jsonObject) throws Exception {
                return jsonObject.getString("id"); //订单明细的id
            }
        });
        SingleOutputStreamOperator<JSONObject> orderJsonWithNegStream = detailIdKeyedStream.process(new KeyedProcessFunction<String, JSONObject, JSONObject>() {
            //状态：  lastOrderJsonObjState  保存该订单明细的上一次数据
            ValueStateDescriptor<JSONObject> lastOrderJsonObjStateDescr = new ValueStateDescriptor<JSONObject>("last_order_json_obj", JSONObject.class);
            ValueState<JSONObject> lastOrderJsonObjState = null;

            @Override
            public void open(Configuration parameters) throws Exception {
                lastOrderJsonObjState = getRuntimeContext().getState(lastOrderJsonObjStateDescr);
            }

            @Override
            public void processElement(JSONObject newOrderJsonObj, KeyedProcessFunction<String, JSONObject, JSONObject>.Context ctx, Collector<JSONObject> out) throws Exception {
                //1 判断是否存在上一次订单明细数据
                //  1.1  如果存在
                //      把状态取出来，把对应的事实度量值，做负数
                //      把旧的反向的数据写入收集器
                //   1.2  如果不存在
                // 2  把 新数据写入状态
                //    把新数据写入收集器
                if (lastOrderJsonObjState.value() != null && lastOrderJsonObjState.value().size() > 0) {
                    JSONObject lastOrderJsonObj = lastOrderJsonObjState.value();
                    BigDecimal couponAmount = lastOrderJsonObj.getBigDecimal("split_coupon_amount") != null ? lastOrderJsonObj.getBigDecimal("split_coupon_amount") : BigDecimal.ZERO;
                    BigDecimal activityAmount = lastOrderJsonObj.getBigDecimal("split_activity_amount") != null ? lastOrderJsonObj.getBigDecimal("split_activity_amount") : BigDecimal.ZERO;
                    BigDecimal totalAmount = lastOrderJsonObj.getBigDecimal("split_total_amount");
                    BigDecimal originalAmount = lastOrderJsonObj.getBigDecimal("split_original_amount");
                    lastOrderJsonObj.put("split_coupon_amount", couponAmount.negate());
                    lastOrderJsonObj.put("split_activity_amount", activityAmount.negate());
                    lastOrderJsonObj.put("split_total_amount", totalAmount.negate());
                    lastOrderJsonObj.put("split_original_amount", originalAmount.negate());

                    out.collect(lastOrderJsonObj);

                }

                lastOrderJsonObjState.update(newOrderJsonObj);
                out.collect(newOrderJsonObj);


            }
        });


        // 2  转换结构
        SingleOutputStreamOperator<TradeSkuOrderBean> orderBeanStream = orderJsonWithNegStream.map(new MapFunction<JSONObject, TradeSkuOrderBean>() {
            @Override
            public TradeSkuOrderBean map(JSONObject jsonObject) throws Exception {
                BigDecimal couponAmount = jsonObject.getBigDecimal("split_coupon_amount")!=null? jsonObject.getBigDecimal("split_coupon_amount"):BigDecimal.ZERO;
                BigDecimal activityAmount = jsonObject.getBigDecimal("split_activity_amount")!=null? jsonObject.getBigDecimal("split_activity_amount"):BigDecimal.ZERO;


                TradeSkuOrderBean tradeSkuOrderBean = TradeSkuOrderBean.builder()
                        .skuId(jsonObject.getString("sku_id"))
                        .couponAmount(couponAmount)
                        .orderAmount(jsonObject.getBigDecimal("split_total_amount"))
                        .originalAmount(jsonObject.getBigDecimal("split_original_amount"))
                        .activityAmount(activityAmount)
                        .skuName(jsonObject.getString("sku_name"))
                        .ts(jsonObject.getLong("ts")).build();

                return tradeSkuOrderBean;
            }
        });




        //orderBeanStream.print();
        // 3  开窗聚合

        //5.1 水位线
        SingleOutputStreamOperator<TradeSkuOrderBean> userLoginBeanWmStream = orderBeanStream.assignTimestampsAndWatermarks(WatermarkStrategy.<TradeSkuOrderBean>forBoundedOutOfOrderness(Duration.ofMillis(1000)).withIdleness(Duration.ofMillis(10000))
                .withTimestampAssigner(new SerializableTimestampAssigner<TradeSkuOrderBean>() {
                    @Override
                    public long extractTimestamp(TradeSkuOrderBean element, long recordTimestamp) {
                        return element.getTs()*1000;
                    }
                }));

        KeyedStream<TradeSkuOrderBean, String> skuIdKeyedStream = userLoginBeanWmStream.keyBy(new KeySelector<TradeSkuOrderBean, String>() {
            @Override
            public String getKey(TradeSkuOrderBean value) throws Exception {
                return value.getSkuId();
            }
        });

        SingleOutputStreamOperator<TradeSkuOrderBean> reduceStream = skuIdKeyedStream.window(TumblingEventTimeWindows.of(Time.seconds(10)))
                .reduce(new ReduceFunction<TradeSkuOrderBean>() {
                    @Override
                    public TradeSkuOrderBean reduce(TradeSkuOrderBean bean1, TradeSkuOrderBean bean2) throws Exception {
                        bean1.setActivityAmount(bean1.getActivityAmount().add(bean2.getActivityAmount()));
                        bean1.setCouponAmount(bean1.getCouponAmount().add(bean2.getCouponAmount()));
                        bean1.setOrderAmount(bean1.getOrderAmount().add(bean2.getOrderAmount()));
                        bean1.setOriginalAmount(bean1.getOriginalAmount().add(bean2.getOriginalAmount()));
                        return bean1;
                    }
                }, new WindowFunction<TradeSkuOrderBean, TradeSkuOrderBean, String, TimeWindow>() {
                    @Override
                    public void apply(String s, TimeWindow window, Iterable<TradeSkuOrderBean> beanItr, Collector<TradeSkuOrderBean> out) throws Exception {
                        String stt = DateFormatUtil.toYmdHms(window.getStart());
                        String edt = DateFormatUtil.toYmdHms(window.getEnd());
                        for (TradeSkuOrderBean tradeSkuOrderBean : beanItr) {
                            tradeSkuOrderBean.setStt(stt);
                            tradeSkuOrderBean.setEdt(edt);
                            tradeSkuOrderBean.setTs(System.currentTimeMillis());
                            out.collect(tradeSkuOrderBean);
                        }
                    }
                });

       // reduceStream.print();

/*
       同步查询方式

        SingleOutputStreamOperator<TradeSkuOrderBean> reduceWithDimStream = reduceStream.map(new MapFunction<TradeSkuOrderBean, TradeSkuOrderBean>() {

            //能不能利用一个状态进行存储 缓存 。。 不行

            @Override
            public TradeSkuOrderBean map(TradeSkuOrderBean tradeSkuOrderBean) throws Exception {
                String skuId = tradeSkuOrderBean.getSkuId();
                JSONObject dimSkuInfo = DimUtil.getDimInfoWithCache("dim_sku_info", skuId);

                tradeSkuOrderBean.setTrademarkId(dimSkuInfo.getString("TM_ID"));
                tradeSkuOrderBean.setSpuId(dimSkuInfo.getString("SPU_ID"));
                tradeSkuOrderBean.setCategory3Id(dimSkuInfo.getString("CATEGORY3_ID"));
                return tradeSkuOrderBean;
            }
        });*/

       // sku_info  spu_info  base_trademark base_category1 .2 .3  ...
        SingleOutputStreamOperator<TradeSkuOrderBean> asyncWithSkuStream = AsyncDataStream.unorderedWait(reduceStream, new AbsDimAsyncFuntion<TradeSkuOrderBean>("dim_sku_info") {
            @Override
            public String getJoinId(TradeSkuOrderBean bean) {
                return bean.getSkuId();
            }

            @Override
            public void join(TradeSkuOrderBean bean, JSONObject dimObject) {
                bean.setTrademarkId(dimObject.getString("tm_id"));
                bean.setSpuId(dimObject.getString("spu_id"));
                bean.setCategory3Id(dimObject.getString("category3_id"));
            }
        } , 60000, TimeUnit.MILLISECONDS, 100);

        //品牌
        SingleOutputStreamOperator<TradeSkuOrderBean> tradeWithSkuTmStream = AsyncDataStream.unorderedWait(asyncWithSkuStream, new AbsDimAsyncFuntion<TradeSkuOrderBean>("dim_base_trademark") {
            @Override
            public String getJoinId(TradeSkuOrderBean bean) {
                return bean.getTrademarkId();
            }

            @Override
            public void join(TradeSkuOrderBean bean, JSONObject dimObject) {
                bean.setTrademarkName(dimObject.getString("tm_name"));

            }
        }, 60, TimeUnit.SECONDS, 100);

        //三级分类
        SingleOutputStreamOperator<TradeSkuOrderBean> tradeWithSkuTmC3Stream = AsyncDataStream.unorderedWait(tradeWithSkuTmStream, new AbsDimAsyncFuntion<TradeSkuOrderBean>("dim_base_category3") {
            @Override
            public String getJoinId(TradeSkuOrderBean bean) {
                return bean.getCategory3Id();
            }

            @Override
            public void join(TradeSkuOrderBean bean, JSONObject dimObject) {
                bean.setCategory2Id(dimObject.getString("category2_id"));
                bean.setCategory3Name(dimObject.getString("name"));
            }
        }, 60, TimeUnit.SECONDS, 100);

        //2级分类
        SingleOutputStreamOperator<TradeSkuOrderBean> tradeWithSkuTmC3C2Stream = AsyncDataStream.unorderedWait(tradeWithSkuTmC3Stream, new AbsDimAsyncFuntion<TradeSkuOrderBean>("dim_base_category2") {
            @Override
            public String getJoinId(TradeSkuOrderBean bean) {
                return bean.getCategory2Id();
            }

            @Override
            public void join(TradeSkuOrderBean bean, JSONObject dimObject) {
                bean.setCategory1Id(dimObject.getString("category1_id"));
                bean.setCategory2Name(dimObject.getString("name"));
            }
        }, 60, TimeUnit.SECONDS, 100);

        //1级分类
        SingleOutputStreamOperator<TradeSkuOrderBean> tradeWithSkuTmC3C2C1Stream = AsyncDataStream.unorderedWait(tradeWithSkuTmC3C2Stream, new AbsDimAsyncFuntion<TradeSkuOrderBean>("dim_base_category1") {
            @Override
            public String getJoinId(TradeSkuOrderBean bean) {
                return bean.getCategory1Id();
            }

            @Override
            public void join(TradeSkuOrderBean bean, JSONObject dimObject) {
                bean.setCategory1Name(dimObject.getString("name"));
            }
        }, 60, TimeUnit.SECONDS, 100);



        //spu
        SingleOutputStreamOperator<TradeSkuOrderBean> tradeWithSkuSpuTmC3C2C1Stream = AsyncDataStream.unorderedWait(tradeWithSkuTmC3C2C1Stream, new AbsDimAsyncFuntion<TradeSkuOrderBean>("dim_spu_info") {
            @Override
            public String getJoinId(TradeSkuOrderBean bean) {
                return bean.getSpuId();
            }

            @Override
            public void join(TradeSkuOrderBean bean, JSONObject dimObject) {
                bean.setSpuName(dimObject.getString("spu_name"));
            }
        }, 60, TimeUnit.SECONDS, 100);

      //  tradeWithSkuSpuTmC3C2C1Stream.print();
        //4  写clickhouse
        tradeWithSkuSpuTmC3C2C1Stream.addSink(ClickhouseUtil.getSink("insert into  dws_trade_sku_order_window values(?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?)"));


        env.execute();

    }
}
